package org.example.nettyrpc;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import org.example.gochannel.JavaGoChannel;
import org.example.nettyrpc.common.IpPortTuple;
import org.example.nettyrpc.common.RequestArgs;
import org.example.nettyrpc.common.ResponseReply;

import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

public class RpcGuest {
    private final EndpointBase endpoint;
    private SendHandler sendHandler = null;
    final IpPortTuple ipPortTuple;
    private volatile boolean hasConnected = false;

    public RpcGuest(EndpointBase endpoint, IpPortTuple ipPortTuple) {
        this.endpoint = endpoint;
        this.ipPortTuple = ipPortTuple;
    }

    public void waitForConnected() {
        while (!hasConnected) {
            Thread.yield();
        }
    }

    public void run() {
        NioEventLoopGroup group = new NioEventLoopGroup();
        sendHandler = new SendHandler();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) {
                            socketChannel.pipeline().addLast(new ObjectDecoder(1024 * 1024,
                                    ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
                            socketChannel.pipeline().addLast(new ObjectEncoder());
                            socketChannel.pipeline().addLast(sendHandler);
                        }
                    });

            ChannelFuture f;
            for (; ; ) {
                try {
                    f = bootstrap.connect(ipPortTuple.getIp(), ipPortTuple.getPort()).sync();
                    hasConnected = true;
                    break;
                } catch (Exception ex) {
                    ex.printStackTrace();
                    Thread.sleep(1000);
                }
            }

            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            group.shutdownGracefully();
        }

    }

    public Object remoteCall(String methodName, Object[] parameters) throws InterruptedException {
        return sendHandler.remoteCall(methodName, parameters);
    }

    private static class SendHandler extends ChannelInboundHandlerAdapter {
        static private final Random random = new Random();
        private ChannelHandlerContext ctx = null;

        private final ConcurrentHashMap<Long, JavaGoChannel<Object>> requestsChannelMap = new ConcurrentHashMap<>();

        protected Object remoteCall(String methodName, Object[] parameters) throws InterruptedException {
            RequestArgs args = new RequestArgs();
            long seq = random.nextLong();
            args.setSeq(seq);
            args.setParameters(parameters);
            args.setMethod(methodName);

            JavaGoChannel<Object> replyChannel = JavaGoChannel.Make();
            requestsChannelMap.put(seq, replyChannel);
            ctx.writeAndFlush(args);

            Object res = replyChannel.out();
            requestsChannelMap.remove(seq);
            return res;
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            this.ctx = ctx;
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof ResponseReply) {
                ResponseReply reply = (ResponseReply) msg;
                if (requestsChannelMap.containsKey(reply.getSeq())) {
                    JavaGoChannel<Object> ch = requestsChannelMap.get(reply.getSeq());
                    ch.clear();
                    ch.in(reply.getRes());
                }
            } else {
                System.out.println("unknown reply");
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }
}
